package com.sprucetec.livechat.service.impl;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.sprucetec.livechat.common.NettyMessageDto;
import com.sprucetec.livechat.constants.MqConstant;
import com.sprucetec.livechat.constants.ParameterConstant;
import com.sprucetec.livechat.model.vo.ResponseJson;
import com.sprucetec.livechat.service.ChatService;
import com.sprucetec.livechat.util.ChatType;
import com.sprucetec.livechat.util.Constant;
import com.sprucetec.mcq.producer.impl.MessageProducer;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.text.MessageFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.DoubleAccumulator;

/**
 * @ClassName ChatServiceImpl
 * @Description:
 * @Author Kenny
 * @Date 2020/4/2
 **/
@Service
@Slf4j
public class ChatServiceImpl implements ChatService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ChatServiceImpl.class);

    @Qualifier("threadPoolTaskExecutor")
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Autowired
    private RedisTemplate redisTemplate;


    @Override
    public void connected(JSONObject param, ChannelHandlerContext ctx) {
        String userId = (String) param.get(ParameterConstant.userId);
        String videoId = String.valueOf(param.get(ParameterConstant.videoId));
        ConcurrentHashMap<String, ChannelHandlerContext> stMap = Constant.onlineUserMap.get(videoId);
        if (stMap == null) {
            ConcurrentHashMap concurrentHashMap = new ConcurrentHashMap();
            concurrentHashMap.put(userId, ctx);
            Constant.onlineUserMap.put(videoId, concurrentHashMap);

        } else {
            stMap.put(userId, ctx);
        }
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.type, ChatType.CONNECTED)
                .setData(ParameterConstant.ctxId, ctx.channel().id().asLongText())
                .toString();
        sendMessage(ctx, responseJson);
        Constant.handlerContextStringMap.put(ctx.channel().id().asLongText(), videoId);
        log.info(MessageFormat.format("userId为 {0} 的用户登记到在线用户表，当前在线人数为：{1}"
                , userId, Constant.onlineUserMap.get(videoId).size()));
    }


    @Override
    public void groupSend(JSONObject param, ChannelHandlerContext ctx) {
        String createUserCode = (String) param.get(ParameterConstant.createUserCode);
        String createUserName = (String) param.get(ParameterConstant.createUserName);
        String content = (String) param.get(ParameterConstant.content);
        String videoId = String.valueOf(param.get(ParameterConstant.videoId));
        String chatType = String.valueOf(param.get(ParameterConstant.chatType));
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.createUserCode, createUserCode)
                .setData(ParameterConstant.createUserName, createUserName)
                .setData(ParameterConstant.content, content)
                .setData(ParameterConstant.videoId, videoId)
                .setData(ParameterConstant.chatType, chatType)
                .setData(ParameterConstant.type, ChatType.GROUP_SENDING)
                .toString();
        Iterator<Entry<String, ChannelHandlerContext>> iterator = Constant.onlineUserMap.get(videoId).entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ChannelHandlerContext> entry = iterator.next();
            ChannelHandlerContext toCtx = Constant.onlineUserMap.get(videoId).get(entry.getKey());
            if (toCtx != null && !entry.getKey().equals(createUserCode)) { // 不发给自己
                sendMessage(toCtx, responseJson);
            }
        }
    }

    @Override
    public void groupSendByMq(NettyMessageDto nettyMessageDto) {
        String videoId = nettyMessageDto.getVideoId();
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.createUserCode, nettyMessageDto.getCreateUserCode())
                .setData(ParameterConstant.createUserName, nettyMessageDto.getCreateUserName())
                .setData(ParameterConstant.content, nettyMessageDto.getContent())
                .setData(ParameterConstant.videoId, videoId)
                .setData(ParameterConstant.chatType, nettyMessageDto.getChatType())
                .setData(ParameterConstant.type, ChatType.GROUP_SENDING)
                .toString();
        log.info("【发送人】{}【内容】{}", nettyMessageDto.getCreateUserName(),
                nettyMessageDto.getContent());
        if (Constant.onlineUserMap.get(videoId) != null) {
            ConcurrentHashMap<String, ChannelHandlerContext> roomCtxMap = Constant.onlineUserMap.get(videoId);
            Iterator<Entry<String, ChannelHandlerContext>> iterator = roomCtxMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, ChannelHandlerContext> entry = iterator.next();
                //ChannelHandlerContext toCtx = roomCtxMap.get(entry.getKey());
                ChannelHandlerContext toCtx = entry.getValue();
                if (toCtx != null && !entry.getKey().equals(nettyMessageDto.getCreateUserCode())) { // 不发给自己
                    sendMessage(toCtx, responseJson);
                }
            }
        }
    }


    /**
     * 直播服务器发送过来的信息
     *
     * @param param
     * @param ctx
     */
    @Override
    public void liveServerSend(JSONObject param, ChannelHandlerContext ctx) {
        String content = (String) param.get(ParameterConstant.content);
        String videoId = String.valueOf(param.get(ParameterConstant.videoId));
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.content, content)
                .setData(ParameterConstant.videoId, videoId)
                .setData(ParameterConstant.type, ChatType.LIVE_SERVER)
                .toString();
        if (Constant.onlineUserMap.get(videoId) == null) return;
        ConcurrentHashMap<String, ChannelHandlerContext> roomCtxMap = Constant.onlineUserMap.get(videoId);
        Iterator<Entry<String, ChannelHandlerContext>> iterator = roomCtxMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ChannelHandlerContext> entry = iterator.next();
            ChannelHandlerContext toCtx = roomCtxMap.get(entry.getKey());
            if (toCtx != null) {
                sendMessage(toCtx, responseJson);
            }
        }
    }


    @Override
    public void liveServerSendByMq(NettyMessageDto messageDto) {
        String content = messageDto.getContent();

        String videoId = messageDto.getVideoId();
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.content, content)
                .setData(ParameterConstant.videoId, videoId)
                .setData(ParameterConstant.type, ChatType.LIVE_SERVER)
                .toString();
        if (Constant.onlineUserMap.get(videoId) == null) return;
        ConcurrentHashMap<String, ChannelHandlerContext> roomCtxMap = Constant.onlineUserMap.get(videoId);
        Iterator<Entry<String, ChannelHandlerContext>> iterator = roomCtxMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ChannelHandlerContext> entry = iterator.next();
            ChannelHandlerContext toCtx = roomCtxMap.get(entry.getKey());
            if (toCtx != null) {
                log.info("MQ 关闭视频=>视频id {0} 内容{1}", videoId, content);
                sendMessage(toCtx, responseJson);
            }
        }
    }

    @Override
    public void removeChannelByMq(NettyMessageDto messageDto) {
        log.info("MQ开始移除视频ID:{} 的相关缓存数据START", messageDto.getVideoId());
        try {
            ConcurrentHashMap<String, ChannelHandlerContext> onlineUserCache = Constant.onlineUserMap.get(messageDto.getVideoId());
            if (onlineUserCache != null) {
                onlineUserCache.forEach((key, channel) -> { // 清空当前视频的channel通道
                    Constant.webSocketHandshakerMap.remove(channel.channel().id().asLongText());
                    Constant.handlerContextStringMap.remove(channel.channel().id().asLongText());
                });
                Constant.onlineUserMap.remove(messageDto.getVideoId()); // 清空当前视频的在线用户
            } else {
                log.info("MQ移除的视频相关缓存数据不存在");
            }
        } catch (Exception ex) {
            log.info("MQ移除视频ID:{} 的相关相关缓存数据异常{}", messageDto.getVideoId(), ex);
        }
        log.info("MQ移除视频ID:{} 的相关缓存数据END", messageDto.getVideoId());
    }

    /*@Override
    public void removeByMq(NettyMessageDto messageDto) {
        ConcurrentHashMap<String, ChannelHandlerContext> stringChannelHandlerContextConcurrentHashMap = Constant.onlineUserMap.get(messageDto.getVideoId());
        if (stringChannelHandlerContextConcurrentHashMap != null) {
            ChannelHandlerContext ctx = stringChannelHandlerContextConcurrentHashMap.get(messageDto.getUserId());
            Iterator<Entry<String, ConcurrentHashMap<String, ChannelHandlerContext>>>
                    iterator = Constant.onlineUserMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, ConcurrentHashMap<String, ChannelHandlerContext>>
                        next = iterator.next();
                ConcurrentHashMap<String, ChannelHandlerContext> value = next.getValue();
                Iterator<Entry<String, ChannelHandlerContext>> it = value.entrySet().iterator();
                while (it.hasNext()) {
                    Entry<String, ChannelHandlerContext> entry = it.next();
                    if (entry.getValue() == ctx) {
                        Constant.webSocketHandshakerMap.remove(ctx.channel().id().asLongText());
                        LOGGER.info(MessageFormat.format("已移除握手实例,用户编号{}，当前握手实例总数为：{0}"
                                , messageDto.getUserId(), Constant.webSocketHandshakerMap.size()));
                        it.remove();
                        LOGGER.info(MessageFormat.format("MCQ 用户离开消息 =>userId为 {0} 的用户已退出聊天，当前在线人数为：{1}"
                                , entry.getKey(), Constant.onlineUserMap.size()));
                        break;
                    }
                }
            }
        }
        LOGGER.info("当前节点不存在该通道");
    }*/

    /**
     * 去Redis中查询在线用户
     *
     * @param videoId
     * @return
     */
    private int getOnlineUserFromRedis(String videoId) {
        return Integer.parseInt(redisTemplate.boundValueOps(MqConstant.ChatMq.onlineUserNum + videoId).get(0, -1));
    }


    @Override
    public void remove(ChannelHandlerContext ctx) {
        Iterator<Entry<String, ConcurrentHashMap<String, ChannelHandlerContext>>>
                iterator = Constant.onlineUserMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Entry<String, ConcurrentHashMap<String, ChannelHandlerContext>>
                    next = iterator.next();
            ConcurrentHashMap<String, ChannelHandlerContext> value = next.getValue();
            Iterator<Entry<String, ChannelHandlerContext>> it = value.entrySet().iterator();
            while (it.hasNext()) {
                Entry<String, ChannelHandlerContext> entry = it.next();
                if (entry.getValue() == ctx) {
                    LOGGER.info("正在移除握手实例...");
                    Constant.webSocketHandshakerMap.remove(ctx.channel().id().asLongText());
                    LOGGER.info(MessageFormat.format("普通remove已移除握手实例，当前握手实例总数为：{0}"
                            , Constant.webSocketHandshakerMap.size()));
                    it.remove();
                    LOGGER.info(MessageFormat.format("userId为 {0} 的用户已退出聊天，当前在线人数为：{1}"
                            , entry.getKey(), Constant.onlineUserMap.size()));
                    break;
                }
            }
        }
    }

    @Autowired
    private MessageProducer mcqMessageProduct;

    /**
     * 用户离开
     *
     * @param ctx
     */
    @Override
    public void removeNotifyByMq(ChannelHandlerContext ctx) {
        String videoId = Constant.handlerContextStringMap.get(ctx.channel().id().asLongText());
        if (StringUtils.isBlank(videoId))
            return;
        Long onlineUserSize = 0L;
        Integer num = (Integer) redisTemplate.boundValueOps(MqConstant.ChatMq.onlineUserNum + videoId).get();
        if (num > 0) {
            onlineUserSize = redisTemplate.boundValueOps(MqConstant.ChatMq.onlineUserNum + videoId).increment(-1);
        } else {
            onlineUserSize = num.longValue();
        }
        NettyMessageDto nettyMessageDto = new NettyMessageDto();
        nettyMessageDto.setVideoId(videoId);
        nettyMessageDto.setOnlineUserNum(onlineUserSize.toString());
        Message message = new Message(MqConstant.TOPIC, MqConstant.VideoDataMq.USER_AWAY,
                MqConstant.VideoDataMq.USER_AWAY_KEY + new Date().getTime(), JSONObject.toJSONString(nettyMessageDto).getBytes(StandardCharsets.UTF_8));
        try {
            mcqMessageProduct.send(message);
        } catch (Exception ex){
            ex.printStackTrace();
        }
    }

    @Override
    public void ping(JSONObject param, ChannelHandlerContext ctx) {
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.type, ChatType.PONG)
                .toString();

        if (Constant.onlineUserMap.get(param.get(ParameterConstant.videoId).toString()) != null) {
            String videoId = param.get(ParameterConstant.videoId).toString();
            Iterator<Entry<String, ChannelHandlerContext>> iterator = Constant.onlineUserMap.get(videoId).entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, ChannelHandlerContext> entry = iterator.next();
                ChannelHandlerContext toCtx = Constant.onlineUserMap.get(videoId).get(entry.getKey());
                if (entry.getKey().equals(param.get(ParameterConstant.createUserCode))) { // 只PONG给自己
                    sendMessage(toCtx, responseJson);
                    break;
                }
            }
        }
    }

    @Override
    public void typeError(ChannelHandlerContext ctx) {
        String responseJson = new ResponseJson()
                .error("该类型不存在！")
                .toString();
        sendMessage(ctx, responseJson);
    }


    private void sendMessage(ChannelHandlerContext ctx, String message) {
        threadPoolTaskExecutor.execute(() -> {
            ctx.channel().writeAndFlush(new TextWebSocketFrame(message));
        });
    }

    /**
     * 用户进入通知在线人
     *
     * @param messageDto
     */
    public void sendOnlineUser(NettyMessageDto messageDto) {
        String videoId = messageDto.getVideoId();
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.onlineUserNum, getOnlineUserFromRedis(videoId))
                .setData(ParameterConstant.type, ChatType.NOTICE)
                .toString();
        sendRoomMessage(videoId, responseJson);
    }

    /**
     * 用户离开通知在线人
     *
     * @param messageDto
     */
    public void sendOnlineUserByUserAway(NettyMessageDto messageDto) {
        String videoId = messageDto.getVideoId();
        String responseJson = new ResponseJson().success()
                .setData(ParameterConstant.onlineUserNum, messageDto.getOnlineUserNum())
                .setData(ParameterConstant.type, ChatType.NOTICE)
                .toString();
        sendRoomMessage(videoId, responseJson);
    }

    private void sendRoomMessage(String videoId, String responseJson) {
        if (Constant.onlineUserMap.get(videoId) != null) {
            ConcurrentHashMap<String, ChannelHandlerContext> strMap = Constant.onlineUserMap.get(videoId);
            Iterator<Entry<String, ChannelHandlerContext>> iterator = strMap.entrySet().iterator();
            while (iterator.hasNext()) {
                Entry<String, ChannelHandlerContext> entry = iterator.next();
                ChannelHandlerContext toCtx = entry.getValue();
                if (toCtx != null) {
                    sendMessage(toCtx, responseJson);
                }
            }
        }
    }
}
